package yanglib.network.tcp;

import yanglib.membership.Membership;
import yanglib.message.Message;
import yanglib.message.MessageDeserializer;
import yanglib.network.MessageHandler;
import yanglib.network.Network;

import java.io.DataInputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 * User: user
 * Date: 9/5/14
 * Time: 2:26 PM
 * To change this template use File | Settings | File Templates.
 */
public class TcpNetwork extends Network{

    private ServerSocket serverSocket = null;
    private boolean running = true;
    public TcpNetwork(Membership membership, MessageHandler msgHandler, MessageDeserializer msgDeserializer, int selfID){
        super(membership, msgHandler, msgDeserializer, selfID);
        int port = membership.getAddress(selfID).getPort();
        System.out.println("binding to port "+port);
        try{
            serverSocket = new ServerSocket(port);
        }
        catch(IOException e){
            e.printStackTrace();
            System.exit(-1);
        }
        new ServerThread(serverSocket).start();
    }

    private class ServerThread extends Thread{
        private ServerSocket serverSocket;

        public ServerThread(ServerSocket serverSocket){
            this.serverSocket = serverSocket;
        }

        public void run(){
            while(running){
                try{
                    Socket socket = serverSocket.accept();
                    System.out.println("accepted "+socket);
                    new ReceiverThread(socket).start();
                }
                catch(IOException e){
                    e.printStackTrace();
                }
            }
        }
    }

    private class ReceiverThread extends Thread{
        private Socket socket;

        public ReceiverThread(Socket socket){
            this.socket = socket;
        }

        public void run(){
            DataInputStream dis = null;
            byte[] buffer = new byte[maxBufferSize];
            try{
                dis = new DataInputStream(socket.getInputStream());
            }
            catch(IOException e){
                e.printStackTrace();
                return;
            }
            while(running){
                try{
                    int size = dis.readInt();
                    System.out.println("receive a message with size "+size);
                    dis.readFully(buffer, 0, size);
                    Message msg = msgDeserializer.deserialize(buffer, 0, size);

                    msgHandler.handleMessage(msg);
                }
                catch(IOException e){
                    e.printStackTrace();
                    return;
                }
            }
        }
    }

    private ConcurrentHashMap<InetSocketAddress, Socket> socketMapping = new ConcurrentHashMap<InetSocketAddress, Socket>();

    @Override
    public void send(byte[] bytes, int offset, int len, InetSocketAddress target) throws IOException {
        if(socketMapping.containsKey(target)){
            socketMapping.get(target).getOutputStream().write(bytes, offset, len);
        }
        else{
            Socket sock = new Socket(target.getAddress(), target.getPort());
            sock.getOutputStream().write(bytes, offset, len);
            socketMapping.put(target, sock);
        }
    }
}
